/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package rabbitmq

import (
	"context"
	"encoding/json"
	"log"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/types"
	drivers "hundsun.com/hsl/hschain/store/mq"
)

const (
	// MqName mq name
	MqName             = "rabbitmq"
	replyTimeoutSecond = 15
)

var conn Connection
var mid int64 // 用作msgId, 全局单调递增

// Service rabbit mq service
type Service struct {
	conn          Connection
	mode          string
	topic         string
	qname         string // queueName
	cname         string // consumerName
	consumer      Consumer
	producers     map[string]*Producer
	replyQname    string // reply queueName
	replyConsumer Consumer
	mu            sync.Mutex
}

func init() {
	drivers.Register(MqName, New)
}

// New 新建实例 mode:worker(task)模式, topic模式
func New(cfg *config.PluginConfig, mode string) (drivers.Client, error) {

	if cfg == nil {
		panic(types.ErrNilRabbitMqConfig)
	}
	conf, err := Map2Config(cfg.Parameters)
	if err != nil {
		panic(err)
	}
	// 为支持Queue的单独关闭，需使用独立conn
	conn, err = NewConn(conf)
	if err != nil {
		panic(err)
	}

	if err := conn.Use(ExchangeDeclare(defaultExchangeName, ExchangeDirect, nil)); err != nil {
		panic(err)
	}

	s := &Service{}
	s.conn = conn
	s.mode = mode
	s.producers = make(map[string]*Producer)
	return s, nil
}

// Send 发送消息 （仅worker模式支持同步发送）
func (s *Service) Send(msg *drivers.Message, sync bool) error {
	if msg == nil {
		return nil
	}

	topic, err := genTopicKey(msg.MainTopic, msg.SubTopic)
	if err != nil {
		return err
	}

	producer := s.loadProducer(topic)

	// sub reply msg
	if sync && drivers.ModeWorker.String() == s.mode {
		replyTopic, err := genReplyTopicKey(msg.MainTopic, msg.SubTopic)
		if err != nil {
			return err
		}
		msg.ID = strconv.FormatInt(atomic.AddInt64(&mid, 1), 10)
		s.replyQname = genReplyQueueName(topic)

		if err := s.subReply(replyTopic, s.replyQname); err != nil {
			return err
		}
		s.replyConsumer = NewConsumer(s.replyQname, "", s.conn)
	}

	m, err := s.encode(msg)
	if err != nil {
		log.Printf("failed to encode msg(%s), err: %v \n", msg.String(), err)
		return err
	}

	if err := (*producer).Publish(m); err != nil {
		log.Println(types.ErrFailedSendMsg.Error())
		return types.ErrFailedSendMsg
	}

	return nil
}

func (s *Service) loadProducer(topic string) *Producer {
	s.mu.Lock()
	defer s.mu.Unlock()

	if _, ok := s.producers[topic]; !ok {
		p := NewProducer(defaultExchangeName, topic, ExchangeDirect, s.conn)
		s.producers[topic] = &p
	}
	return s.producers[topic]
}

// SendTimeout send timeout
func (s *Service) SendTimeout(msg *drivers.Message, sync bool, timeout time.Duration) error {
	if timeout <= 0 {
		return s.Send(msg, sync)
	}

	var err error
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	go func(ctx context.Context) {
		err = s.Send(msg, sync)
	}(ctx)

	select {
	case <-ctx.Done():
		return err
	case <-time.After(timeout):
		return types.ErrSendTimeout
	}
}

// Wait 等待应答消息(req--rsp)
func (s *Service) Wait(msg *drivers.Message) (*drivers.Message, error) {
	if len(msg.ID) == 0 {
		return nil, types.ErrNilMsgID
	}

	if s.replyConsumer == nil {
		return nil, types.ErrNilReplyConsumer
	}

	msgs, err := s.replyConsumer.Recv()
	if err != nil {
		return nil, err
	}

	for d := range msgs {
		if msg.ID == d.CorrelationId {
			if err := d.Ack(false); err != nil {
				log.Printf("failed to ack msg-%s, err: %v \n", msg.ID, err)
				return nil, err
			}

			m, err := s.decode(d.Body)
			if err != nil {
				log.Printf("failed to decode msg-%s, err: %v \n", msg.ID, err)
				return nil, err
			}

			return m, nil
		}
	}

	return nil, types.ErrNilReplyMsg
}

// WaitTimeout wait timeout
func (s *Service) WaitTimeout(msg *drivers.Message, timeout time.Duration) (*drivers.Message, error) {
	if timeout <= 0 {
		return s.Wait(msg)
	}

	var m *drivers.Message
	var err error
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	go func(ctx context.Context) {
		m, err = s.Wait(msg)
	}(ctx)

	select {
	case <-ctx.Done():
		return m, err
	case <-time.After(timeout):
		return nil, types.ErrWaitTimeout
	}
}

// Recv 接收订阅消息
func (s *Service) Recv() chan *drivers.Message {
	s.cname = ""
	s.consumer = NewConsumer(s.qname, s.cname, s.conn)

	// buffer with the same capacity 5
	c := make(chan *drivers.Message, 5)
	go func() {
		msgs, _ := s.consumer.Recv()
		for msg := range msgs {
			if err := msg.Ack(false); err != nil {
				log.Printf("failed to ack msg-%s, err: %v \n", msg.MessageId, err)
				continue
			}

			m, err := s.decode(msg.Body)
			if err != nil {
				log.Printf("failed to decode msg-%v, err: %v \n", m, err)
				continue
			}

			if len(m.ID) > 0 {
				m.ChReply = make(chan *drivers.Message, 1)
				go func(m *drivers.Message) {
					topic, err := genReplyTopicKey(m.MainTopic, m.SubTopic)
					if err != nil {
						log.Printf("failed to gen topic key for reply msg-%s, err: %v \n", m.ID, err)
						return
					}

					select {
					case i := <-m.ChReply:
						input, err := s.encode(i)
						if err != nil {
							log.Printf("failed to encode reply msg-%s, err: %v \n", m.ID, err)
							return
						}
						producer := s.loadProducer(topic)
						err = (*producer).PublishWithReply(m.ID, topic, input)
						if err != nil {
							log.Printf("failed to reply msg-%s, err: %v \n", m.ID, err)
							return
						}
					case <-time.After(time.Second * replyTimeoutSecond):
						// 强制超时退出
						return
					}

				}(m)
			}

			// log.Printf("ack msg: %v", m.Data)

			c <- m
		}
	}()
	return c
}

// Sub 订阅消息
func (s *Service) Sub(mainTopic, subTopic string) error {
	topic, err := genTopicKey(mainTopic, subTopic)
	if err != nil {
		return err
	}

	s.topic = topic

	switch s.mode {
	case drivers.ModeTopic.String():
		s.qname = genQueueName(topic)
	case drivers.ModeWorker.String():
		s.qname = s.topic
	default:
		return types.ErrWrongMqMode
	}

	if err := s.conn.Use(QueueDeclare(s.qname, nil)); err != nil {
		return err
	}
	if err := s.conn.Use(QueueBind(s.qname, s.topic, defaultExchangeName, false, nil)); err != nil {
		return err
	}

	return s.conn.Run()
}

func (s *Service) subReply(topic, replyQname string) error {
	if err := s.conn.Use(QueueDeclare(replyQname, nil)); err != nil {
		return err
	}
	if err := s.conn.Use(QueueBind(replyQname, topic, defaultExchangeName, false, nil)); err != nil {
		return err
	}

	return s.conn.Run()
}

// Close 关闭client
func (s *Service) Close() {
	if s.consumer != nil {
		s.consumer.Close()
	}
}

// NewMessage 新建消息
func (s *Service) NewMessage(mainTopic, subTopic string, ty int64, data interface{}) *drivers.Message {
	return &drivers.Message{MainTopic: mainTopic, SubTopic: subTopic, Ty: ty, Data: data}
}

func (s *Service) encode(msg *drivers.Message) ([]byte, error) {
	return json.Marshal(*msg)
}

func (s *Service) decode(b []byte) (*drivers.Message, error) {
	var m drivers.Message
	if err := json.Unmarshal(b, &m); err != nil {
		return &m, err
	}
	return &m, nil
}
